package com.study;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * @author sjw
 * @description
 * @date 2019-08-20 上午10:11
 */
public class MapReduce {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Integer> dataStream = env.addSource(new SourceFunction<Integer>() {
            //private static final long serialVersionUID = 1L;

            //private volatile boolean isRunning = true;
            @Override
            public void run(SourceContext<Integer> sourceContext) throws Exception {
                for (int i = 0; i < 18; i++) {
                    //sourceContext.collect((int) (System.currentTimeMillis() % 100));
                    sourceContext.collect(i%10);
                    //System.out.println(i+"------"+i%10);
                    Thread.sleep(100);
                }
            }

            @Override
            public void cancel() {

            }
        });


        DataStream<IntCount> intCountDataStream=dataStream.flatMap(new FlatMapFunction<Integer, IntCount>() {
            @Override
            public void flatMap(Integer integer, Collector<IntCount> collector) throws Exception {
                collector.collect(new IntCount(integer,1));
            }
        }).keyBy("value").sum("count");
//                countWindow(2)
//                .reduce(new ReduceFunction<IntCount>() {
//            @Override
//            public IntCount reduce(IntCount intCount, IntCount t1) throws Exception {
//                return new IntCount(intCount.value,intCount.count+t1.value);
//            }
//        });

        intCountDataStream.print().setParallelism(1);


        //dataStream.sp

//        SingleOutputStreamOperator<Integer> map=dataStream.map(new MapFunction<Integer, Integer>() {
//            @Override
//            public Integer map(Integer integer) throws Exception {
//                return integer;
//            }
//        });
//        map.print();

//        SingleOutputStreamOperator<Integer> flatMap=dataStream.flatMap(new FlatMapFunction<Integer, Integer>() {
//            @Override
//            public void flatMap(Integer integer, Collector<Integer> collector) throws Exception {
//                if(integer%2==0) collector.collect(integer);
//            }
//        });
//
//        flatMap.print();

//        SingleOutputStreamOperator<Integer> filter=dataStream.filter(new FilterFunction<Integer>() {
//            @Override
//            public boolean filter(Integer integer) throws Exception {
//                return integer>30;
//            }
//        });
//        filter.print();


//        dataStream.flatMap(new FlatMapFunction<Integer, Integer>() {
//            @Override
//            public void flatMap(Integer integer, Collector<Integer> collector) throws Exception {
//                collector.collect(integer);
//            }
//        }).keyBy(new KeySelector<Integer, Integer>() {
//            @Override
//            public Integer getKey(Integer integer) throws Exception {
//                return integer % 5;
//            }
//        }).sum(0).print();

//        KeyedStream<Integer,Integer> keyBy=dataStream.keyBy(new KeySelector<Integer, Integer>() {
//            @Override
//            public Integer getKey(Integer integer) throws Exception {
//                return integer%5;
//            }
//        });
//        keyBy.sum(0).print();
//        keyBy.reduce(new ReduceFunction<Integer>() {
//            @Override
//            public Integer reduce(Integer integer, Integer t1) throws Exception {
//                return integer+t1;
//            }
//        }).print();


        env.execute();

    }


    public static class IntCount {
        public Integer value;
        public int count;

        public IntCount() {
        }

        public IntCount(Integer value, int count) {
            this.value = value;
            this.count = count;
        }

        @Override
        public String toString() {
            return value + " : " + count;
        }

    }

}
